package com.group2.edu.realtime.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.group2.edu.realtime.common.base.BaseApp;
import com.group2.edu.realtime.common.bean.DwsLearnChapterPlayWindowBean;
import com.group2.edu.realtime.common.constant.Constant;
import com.group2.edu.realtime.common.function.DimAsyncFunction;
import com.group2.edu.realtime.common.util.DateFormatUtil;
import com.group2.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;


/**
 *
 * 需要启动的进程：zk、kafka、flume、doris、DwdBaseLog、DwdLearningPlay，hbase,redis
 */
public class DwsLearnChapterPlayWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsLearnChapterPlayWindow().start(
                10024,
                4,
                "dws_learn_chapter_play_window",
                Constant.TOPIC_DWD_LEARNING_PLAY
        );

    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> playSourceStream) {
        SingleOutputStreamOperator<DwsLearnChapterPlayWindowBean> beanStream = playSourceStream.map(new MapFunction<String, DwsLearnChapterPlayWindowBean>() {
            @Override
            public DwsLearnChapterPlayWindowBean map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);

                return DwsLearnChapterPlayWindowBean.builder()
                        .videoId(jsonObject.getString("videoId"))
                        .userId(jsonObject.getString("userId"))
                        .playTotalSec(jsonObject.getLong("playSec"))
                        .playCount(1L)
                        .ts(jsonObject.getLong("ts"))
                        .build();
            }
        });
        //beanStream.print();
        KeyedStream<DwsLearnChapterPlayWindowBean, String> keyedStream = beanStream.keyBy(new KeySelector<DwsLearnChapterPlayWindowBean, String>() {
            @Override
            public String getKey(DwsLearnChapterPlayWindowBean value) throws Exception {
                return value.getUserId();
            }
        });

        SingleOutputStreamOperator<DwsLearnChapterPlayWindowBean> uuBeanStream = keyedStream.process(new KeyedProcessFunction<String, DwsLearnChapterPlayWindowBean, DwsLearnChapterPlayWindowBean>() {
            @Override
            public void processElement(DwsLearnChapterPlayWindowBean bean, KeyedProcessFunction<String, DwsLearnChapterPlayWindowBean, DwsLearnChapterPlayWindowBean>.Context context, Collector<DwsLearnChapterPlayWindowBean> out) throws Exception {
                String lastDt = lastPlayDtState.value();
                String curDt = DateFormatUtil.tsToDate(bean.getTs());
                if (lastDt == null || lastDt.compareTo(curDt) < 0) {
                    bean.setPlayUuCount(1L);
                    lastPlayDtState.update(curDt);
                } else {
                    bean.setPlayUuCount(0L);
                }
                out.collect(bean);
            }

            ValueState<String> lastPlayDtState = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                ValueStateDescriptor<String> lastPlayDtStateDesc = new ValueStateDescriptor<>("last_play_dt_state", String.class);
                lastPlayDtStateDesc.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                lastPlayDtState = getRuntimeContext().getState(lastPlayDtStateDesc);
            }


        });
        //uuBeanStream.print();
        SingleOutputStreamOperator<DwsLearnChapterPlayWindowBean> withWaterMarkStream = uuBeanStream.assignTimestampsAndWatermarks(WatermarkStrategy.<DwsLearnChapterPlayWindowBean>forBoundedOutOfOrderness(Duration.ofSeconds(5L)).withTimestampAssigner(new SerializableTimestampAssigner<DwsLearnChapterPlayWindowBean>() {
            @Override
            public long extractTimestamp(DwsLearnChapterPlayWindowBean element, long recordTimestamp) {
                return element.getTs();
            }
        }));

        SingleOutputStreamOperator<DwsLearnChapterPlayWindowBean> reduceStream = withWaterMarkStream.keyBy(DwsLearnChapterPlayWindowBean::getVideoId)
                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10L)))
                .reduce(new ReduceFunction<DwsLearnChapterPlayWindowBean>() {
                    @Override
                    public DwsLearnChapterPlayWindowBean reduce(DwsLearnChapterPlayWindowBean value1, DwsLearnChapterPlayWindowBean value2) throws Exception {
                        // 相同组的数据度量值累加
                        value1.setPlayCount(value1.getPlayCount() + value2.getPlayCount());
                        value1.setPlayTotalSec(value1.getPlayTotalSec() + value2.getPlayTotalSec());
                        value1.setPlayUuCount(value1.getPlayUuCount() + value2.getPlayUuCount());
                        return value1;
                    }
                }, new ProcessWindowFunction<DwsLearnChapterPlayWindowBean, DwsLearnChapterPlayWindowBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<DwsLearnChapterPlayWindowBean> elements, Collector<DwsLearnChapterPlayWindowBean> out) throws Exception {
                        String stt = DateFormatUtil.tsToDateTime(context.window().getStart());
                        String edt = DateFormatUtil.tsToDateTime(context.window().getEnd());
                        String curDate = DateFormatUtil.tsToDate(context.window().getStart());
                        for (DwsLearnChapterPlayWindowBean element : elements) {
                            element.setStt(stt);
                            element.setEdt(edt);
                            element.setCurDate(curDate);
                            element.setTs(System.currentTimeMillis());
                            out.collect(element);
                        }
                    }
                });
        //reduceStream.print();
        SingleOutputStreamOperator<DwsLearnChapterPlayWindowBean> withProvinceDS = AsyncDataStream.unorderedWait(
                reduceStream,
                new DimAsyncFunction<DwsLearnChapterPlayWindowBean>() {
                    @Override
                    public void addDims(DwsLearnChapterPlayWindowBean obj, JSONObject dimJsonObj) {
                        obj.setChapterId(dimJsonObj.getString("chapter_id"));

                    }

                    @Override
                    public String getTableName() {
                        return "dim_video_info";
                    }

                    @Override
                    public String getRowKey(DwsLearnChapterPlayWindowBean obj) {
                        return obj.getVideoId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );


        SingleOutputStreamOperator<DwsLearnChapterPlayWindowBean> withwindow = AsyncDataStream.unorderedWait(
                withProvinceDS,
                new DimAsyncFunction<DwsLearnChapterPlayWindowBean>() {
                    @Override
                    public void addDims(DwsLearnChapterPlayWindowBean obj, JSONObject dimJsonObj) {
                        obj.setChapterName(dimJsonObj.getString("chapter_name"));

                    }

                    @Override
                    public String getTableName() {
                        return "dim_chapter_info";
                    }

                    @Override
                    public String getRowKey(DwsLearnChapterPlayWindowBean obj) {
                        return obj.getChapterId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );

        withwindow.print();

        withwindow
                .map(
                        new MapFunction<DwsLearnChapterPlayWindowBean, String>() {
                            @Override
                            public String map(DwsLearnChapterPlayWindowBean bean) throws Exception {
                                SerializeConfig config = new SerializeConfig();
                                config.setPropertyNamingStrategy(PropertyNamingStrategy.SnakeCase);
                                String jsonStr = JSON.toJSONString(bean, config);
                                return jsonStr;
                            }
                        }
                )
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_learn_chapter_play_window"));









    }
}
